"""
Case Type   : 协议兼容
Case Name   : 一次为多个表创建预备语句
Create At   : 2024.09.23
Owner       : @zhanghuan96
Description :
    1、创建兼容b库并查询插件是否加载成功
    2、创建用户
    3、caching_sha2认证
    4、查询系统未使用端口
    5、设置协议兼容guc参数
    6、创建预备语句含2个表
    7、结合查询计划执行预备语句
    8、创建预备语句含?占位符
    9、加using执行
    10、结合查询计划执行预备语句
    11、清理环境
Expect      :
    1、成功，插件查询成功
    2、成功
    3、成功
    4、成功
    5、成功
    6、成功
    7、成功
    8、成功
    9、成功
    10、成功
    11、成功
History     :
"""
import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonMySQL import CommonMySQL
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class CompatibilityTest01(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.common = Common()
        self.sh_pri = CommonSH('PrimaryDbUser')
        self.pri_node = Node('PrimaryDbUser')
        self.my_node = Node('mysql8')
        self.constant = Constant()
        self.db_name = 'db_protocol_0045'
        self.u_name = 'u_protocol_0045'
        self.t_name = 't_protocol_0045'
        self.conf_path = os.path.join(macro.DB_INSTANCE_PATH,
                                      'postgresql.conf')
        self.default_value1 = self.common.show_param('enable_dolphin_proto')
        self.default_value2 = self.common.show_param('dolphin_server_port')
        self.exec_sql1 = f'''drop table if exists {self.t_name};
        create table {self.t_name}(a int, b numeric) ;
        drop table if exists {self.t_name}_01;
        create table {self.t_name}_01(a int, b int) ;
        insert into {self.t_name} values(1, 1),(1, 1),(1, 1);
        insert into {self.t_name}_01 values(1, 1),(1, 1),(1, 1);
        set dolphin.b_compatibility_mode to on;
        prepare p_prepare0011_01 from 'select * from {self.t_name}, \
        {self.t_name}_01 where {self.t_name}.a = {self.t_name}_01.a';
        execute p_prepare0011_01;'''
        self.exec_sql2 = f'''set dolphin.b_compatibility_mode to on;
         prepare p_prepare0011_01 from 'select * from {self.t_name}, \
        {self.t_name}_01 where {self.t_name}.a = {self.t_name}_01.a';
        explain (costs off, verbose on) execute p_prepare0011_01;'''
        self.exec_sql3 = f'''prepare p_prepare0011_02(int) from 'select * \
        from {self.t_name}, \
        {self.t_name}_01 where {self.t_name}.a = {self.t_name}_01.a and \
        {self.t_name}.b = ?';
        execute p_prepare0011_02(1);'''
        self.exec_sql4 = f'''prepare p_prepare0011_02(int) from 'select * \
        from {self.t_name}, \
        {self.t_name}_01 where {self.t_name}.a = {self.t_name}_01.a and \
        {self.t_name}.b = ?';
        execute p_prepare0011_02 using 1;'''
        self.exec_sql5 = f'''prepare p_prepare0011_02(int) from 'select * \
        from {self.t_name}, \
        {self.t_name}_01 where {self.t_name}.a = {self.t_name}_01.a and \
        {self.t_name}.b = ?';
        explain (costs off, verbose on) execute p_prepare0011_02(1);'''

    def test_create_function(self):
        text = '-----step1:创建兼容b库并查询插件;expect:成功-----'
        self.log.info(text)
        sql_cmd = (f"drop database if exists {self.db_name};"
                   f"create database {self.db_name} dbcompatibility ='B';")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertEqual(1,
                         sql_res.count(self.constant.CREATE_DATABASE_SUCCESS),
                         '兼容B库创建失败')
        sql_cmd1 = "select extname from pg_extension where extname ='dolphin';"
        self.log.info(sql_cmd1)
        sql_res1 = self.sh_pri.execut_db_sql(sql_cmd1, dbname=self.db_name)
        self.log.info(sql_res1)
        self.assertIn('dolphin', sql_res1, '查询失败')

        text = '---step2:创建用户;expect:成功---'
        self.log.info(text)
        sql_cmd2 = (f"drop user if exists {self.u_name};"
                    f"create user {self.u_name} with password "
                    f"'{macro.COMMON_PASSWD}';"
                    f"grant all privileges to {self.u_name};")
        self.log.info(sql_cmd2)
        sql_res2 = self.sh_pri.execut_db_sql(sql_cmd2, dbname=self.db_name)
        self.log.info(sql_res2)
        self.assertIn('CREATE ROLE', sql_res2, '创建失败')
        self.assertIn('ALTER ROLE', sql_res2, '创建失败')

        text = '---step3:caching_sha2认证;expect:成功---'
        self.log.info(text)
        sql_cmd = (f"select set_caching_sha2_password('{self.u_name}', "
                   f"'{macro.COMMON_PASSWD}', '');")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('(1 row)', sql_res, '创建失败')

        text = '----step4:查询系统未使用端口;expect:成功----'
        self.log.info(text)
        port = self.common.get_not_used_port(self.pri_node)
        self.log.info(port)
        self.assertNotEqual(0, port, '执行失败:' + text)

        text = '-step5:设置协议兼容guc参数;expect:成功-'
        self.log.info(text)
        msg = self.sh_pri.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        'enable_dolphin_proto = on')
        self.log.info(msg)
        self.assertTrue(msg)
        msg = self.sh_pri.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f'dolphin_server_port ={port}')
        self.log.info(msg)
        self.assertTrue(msg)
        sed_cmd = (f"sed -i '$a dolphin.default_database_name={self.db_name}' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

        text = '-step6:创建预备语句含2个表;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql1, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('\na\tb\ta\tb\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1\t1\t1'
                      '\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1'
                      '\t1\t1\n1\t1\t1\t1', c_res)

        text = '-step7:结合查询计划执行预备语句;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql2, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('Output: t_protocol_0045.a, t_protocol_0045.b, '
                      't_protocol_0045_01.a, t_protocol_0045_01.b', c_res)

        text = '-step8:创建预备语句含?占位符;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql3, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('\na\tb\ta\tb\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1\t1\t1'
                      '\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1'
                      '\t1\t1\n1\t1\t1\t1', c_res)

        text = '-step9:加using执行;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql4, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('\na\tb\ta\tb\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1\t1\t1'
                      '\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1\t1\t1\n1\t1'
                      '\t1\t1\n1\t1\t1\t1', c_res)

        text = '-step10:结合查询计划执行预备语句;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql5, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('QUERY PLAN', c_res)

    def tearDown(self):
        text = '-----step11:清理环境;expect:成功-----'
        self.log.info(text)
        drop_cmd = f"drop database if exists {self.db_name};"
        self.log.info(drop_cmd)
        drop_res = self.sh_pri.execut_db_sql(drop_cmd, dbname='postgres')
        self.log.info(drop_res)
        drop_cmd2 = f"drop user if exists {self.u_name};"
        self.log.info(drop_cmd2)
        drop_res2 = self.sh_pri.execut_db_sql(drop_cmd2)
        self.log.info(drop_res2)

        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'enable_dolphin_proto = '
                                            f'{self.default_value1}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)
        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'dolphin_server_port ='
                                            f'{self.default_value2}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)

        sed_cmd = (f"sed -i '/dolphin.default_database_name={self.db_name}/d' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.log.info(status)
        self.assertTrue("Degraded" in status or "Normal" in status)
        self.assertIn('DROP ROLE', drop_res2, '清理环境失败')
        self.assertEqual(1,
                         drop_res.count(self.constant.DROP_DATABASE_SUCCESS),
                         '清理环境失败')
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
